package com.bwie.service.impl;

import cn.hutool.core.util.RandomUtil;
import cn.hutool.crypto.digest.DigestUtil;
import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.bwie.domain.TbLog;
import com.bwie.domain.TbUser;
import com.bwie.es.EsLog;
import com.bwie.mapper.TbLogMapper;
import com.bwie.mapper.TbUserMapper;
import com.bwie.service.TbUserService;
import com.bwie.utils.MobileMessageUtil;
import com.bwie.utils.R;
import com.bwie.vo.LoginInfoVo;
import com.bwie.vo.MessageVo;
import com.bwie.vo.PageInfoVo;
import io.jsonwebtoken.Jwts;
import io.jsonwebtoken.SignatureAlgorithm;
import org.apache.commons.lang3.StringUtils;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.fetch.subphase.highlight.HighlightBuilder;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.domain.PageRequest;
import org.springframework.data.elasticsearch.core.ElasticsearchRestTemplate;
import org.springframework.data.elasticsearch.core.SearchHit;
import org.springframework.data.elasticsearch.core.SearchHits;
import org.springframework.data.elasticsearch.core.query.NativeSearchQueryBuilder;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Service;
import org.springframework.util.concurrent.FailureCallback;
import org.springframework.util.concurrent.SuccessCallback;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.TimeUnit;

/**
* @author andy
* @description 针对表【tb_user(用户表)】的数据库操作Service实现
* @createDate 2024-09-24 10:34:14
*/
@Service
public class TbUserServiceImpl extends ServiceImpl<TbUserMapper, TbUser>
    implements TbUserService{

    @Autowired
    KafkaTemplate kafkaTemplate;

    @Autowired
    StringRedisTemplate stringRedisTemplate;

    @Autowired
    TbUserMapper tbUserMapper;

    @Autowired
    TbLogMapper tbLogMapper;

    @Autowired
    ElasticsearchRestTemplate elasticsearchRestTemplate;

    @Override
    public R userLogin(LoginInfoVo loginInfoVo) {
        //-- 校验验证码
        String code = stringRedisTemplate.opsForValue().get(loginInfoVo.getUserMobile());
        if(code == null) {
            return R.ERROR(500, "验证码已过期");
        }
        if(!code.equals(loginInfoVo.getUserCode())) {
            return R.ERROR(500, "验证码错误");
        }

        //-- 获取用户信息
        TbUser tbUser = tbUserMapper.selectOne(new QueryWrapper<TbUser>().lambda().eq(TbUser::getUserName, loginInfoVo.getUserName()));
        if(tbUser == null) {
            return R.ERROR(500, "用户不存在");
        }

        //-- 校验密码
        String bcrypt = DigestUtil.bcrypt(loginInfoVo.getUserPass());
        System.out.println("加密密码为："+bcrypt);

        boolean check = DigestUtil.bcryptCheck(loginInfoVo.getUserPass(), tbUser.getUserPass());
        if(!check) {
            return R.ERROR(500, "密码错误");
        }

        //-- 生成token，并存入缓存
        String token = Jwts.builder()
                .signWith(SignatureAlgorithm.HS256, "123456")
                .claim("userId", tbUser.getUserId())
                .claim("userName", tbUser.getUserName())
                .compact();
        String tokenKey = "token-" + tbUser.getUserId();
        stringRedisTemplate.opsForValue().set(tokenKey, token, 30, TimeUnit.MINUTES);

        //-- 返回用户信息
        tbUser.setUserPass(null);
        tbUser.setToken(token);
        return R.OK(tbUser);
    }

    @Override
    public R sendCode(String userMobile) {
        // 通过kafka发送手机号

        MessageVo messageVo = new MessageVo();

        // 生成消息id，防止消息重复
        String msgId = UUID.randomUUID().toString();
        stringRedisTemplate.opsForValue().set(msgId, msgId);

        // 生成消息
        messageVo.setMsgId(msgId);
        messageVo.setMsgBody(userMobile);

        // 发送消息
        kafkaTemplate.send("codeTopic", JSON.toJSONString(messageVo)).addCallback(
                new SuccessCallback() {
                    @Override
                    public void onSuccess(Object o) {
                        System.out.println("发送成功");
                    }
                },
                new FailureCallback() {
                    @Override
                    public void onFailure(Throwable throwable) {
                        System.out.println("发送失败，再次发送消息");
                        kafkaTemplate.send("codeTopic", JSON.toJSONString(messageVo));
                    }
                }
        );

        return R.OK();
    }

    @Override
    public R getSysLogByPage(PageInfoVo pageInfoVo) {
        //-- 构造分页
        int pageNum = 0;
        if(pageInfoVo.getPageNum() > 0) {
            pageNum = pageInfoVo.getPageNum() - 1;
        }
        PageRequest pageRequest = PageRequest.of(pageNum, pageInfoVo.getPageSize());

        //-- 构造查询条件
        NativeSearchQueryBuilder nativeSearchQueryBuilder = new NativeSearchQueryBuilder();

        BoolQueryBuilder boolQueryBuilder = new BoolQueryBuilder();

        if(!StringUtils.isEmpty(pageInfoVo.getKeyWord())) {
            boolQueryBuilder.must(QueryBuilders.matchQuery("logContent", pageInfoVo.getKeyWord()));
        }

        HighlightBuilder highlightBuilder = new HighlightBuilder();
        highlightBuilder.field("logContent").preTags("<font color='red'>").postTags("</font>");

        nativeSearchQueryBuilder.withPageable(pageRequest)
                .withQuery(boolQueryBuilder)
                .withHighlightBuilder(highlightBuilder);

        //-- 查询
        SearchHits<EsLog> searchHits = elasticsearchRestTemplate.search(nativeSearchQueryBuilder.build(), EsLog.class);

        //-- 解析查询结果
        long total = searchHits.getTotalHits();
        ArrayList<EsLog> records = new ArrayList<>();

        for (SearchHit<EsLog> searchHit : searchHits.getSearchHits()) {
            EsLog esLog = searchHit.getContent();

            // 获取高亮字段
            List<String> logContents = searchHit.getHighlightField("logContent");
            if(logContents != null && logContents.size() > 0) {
                String logContent = logContents.get(0);
                esLog.setLogContent(logContent);
            }

            records.add(esLog);
        }

        //-- 返回数据
        HashMap<String, Object> map = new HashMap<>();
        map.put("total", total);
        map.put("records", records);

        return R.OK(map);
    }

    @Override
    public R getUserById(Integer userId) {
        TbUser tbUser = tbUserMapper.selectById(userId);
        if(tbUser != null) {
            return R.OK(tbUser);
        }
        return R.ERROR(404, "用户不存在");
    }

    @KafkaListener(topics = "codeTopic")
    public void recvCodeMessage(String message, Acknowledgment ack) {
        System.out.println("Kafka接收到消息：" + message);

        //-- 接收并解析消息
        MessageVo messageVo = JSON.parseObject(message, MessageVo.class);
        String userMobile = messageVo.getMsgBody();

        //-- 判断消息是否重复
        String msgId = messageVo.getMsgId();
        Boolean hasKey = stringRedisTemplate.hasKey(msgId);
        if(!hasKey) {
            System.out.println("消息已经重复了，删除消息");
            ack.acknowledge();
            return;
        }

        //-- 消费消息
        // 生成验证码
        String code = RandomUtil.randomNumbers(4);
        System.out.println("验证码：" + code);
        // 发送验证码
        Boolean ok = MobileMessageUtil.sendMobileMessage(userMobile, code);
        if(!ok) {
            System.out.println("发送短信失败");
            MobileMessageUtil.sendMobileMessage(userMobile, code);
        }
        // 验证码放入缓存（过期时间为5分钟）
        stringRedisTemplate.opsForValue().set(userMobile, code, 5, TimeUnit.MINUTES);

        //-- 消息确认，防止消息丢失
        ack.acknowledge();

        //-- 删除缓存中的ID，防止消息重复消费
        stringRedisTemplate.delete(msgId);
    }
}




